package com.hz.cheetah.client;

import com.hz.cheetah.common.Constants;
import com.hz.cheetah.common.bean.RpcRequest;
import com.hz.cheetah.common.bean.RpcResponse;
import com.hz.cheetah.common.codec.RpcDecoder;
import com.hz.cheetah.common.codec.RpcEncoder;
import com.hz.cheetah.hub.protocol.CheetahURL;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * netty客户端，用户远程调用服务
 *
 * @author yangchuan
 * @since 1.0.0
 */
public class CheetahClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(CheetahClient.class);

    private final CheetahURL url;

    private volatile boolean connected = false;
    private volatile Bootstrap bootstrap = null;

    private Lock readyLock = new ReentrantLock();
    private Condition readied = readyLock.newCondition();

    EventLoopGroup workerGroup = new NioEventLoopGroup(Constants.DEFAULT_IO_THREADS);


    private volatile Channel channel;
    private volatile CChannel cChannel;


    /**
     * 最后接收到的消息时间
     */
    public long lastmsgtime;
    /**
     * 心跳测试次数，如果超过三次就关闭客户端
     */
    public int tryitnum;

    public CheetahClient(CheetahURL url) {
        this.url = url;
        open();
    }

    /**
     *
     */
    public void open() {
        // 创建并初始化 Netty 客户端 Bootstrap 对象
        bootstrap = new Bootstrap();
        bootstrap.group(workerGroup)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .option(ChannelOption.SO_SNDBUF, 1024)
                .option(ChannelOption.SO_RCVBUF, 1024)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Constants.DEFAULT_CONNECT_TIMEOUT)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel channel) throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast(new RpcEncoder(RpcRequest.class)) // 编码 RPC 请求
                                .addLast(new RpcDecoder(RpcResponse.class)) // 解码 RPC 响应
                                .addLast(new ChannelHandler(url.getAdress())); // 处理 RPC 响应
                    }
                });

    }

    public void connection() {
        //不能用线程队列，client在关闭前会堵塞，队列也会堵塞
        new ChannelTask().start();
    }

    public void close() {
        if(this.channel != null) {
            this.channel.close();
        }
    }

    class ChannelTask extends Thread {

        private volatile boolean flag = false;

        private void lock(){
            readyLock.lock();
            flag = true;
        }

        private void unlock(){
            readied.signalAll();
            readyLock.unlock();
            flag = false;
        }

        @Override
        public void run() {
            try {
                if (!connected) {
                    lock();
                    if (!connected) {
                        // 连接 RPC 服务器
                        ChannelFuture future = bootstrap.connect(url.getRemoteAddress()).sync();
                        Channel channel = future.channel();
                        CheetahClient.this.channel = channel;
                        initChannel(channel);
                        connected = true;
                        resetLasttime();
                        unlock();
                        channel.closeFuture().sync();
                        connected = false;
                    }
                }
                // 返回 RPC 响应对象
            } catch (Exception e) {
                connected = false;
                LOGGER.error(CheetahClient.class.toString(), e);
            } finally {
                if (flag) {
                    unlock();
                } else {
                    LOGGER.info("client closed, service address is {}", url.getAdress());
                }
            }
        }
    }

    protected void initChannel(Channel channel) {
        this.cChannel = new CChannel();
        this.cChannel.setLocalAddress(channel.localAddress());
        this.cChannel.setRemoteAddress(channel.remoteAddress());
    }


    public RpcResponse send(RpcRequest request) throws Exception {

        if (!connected) {
            //if (readyLock.tryLock()) {
            /**
             * readyLock 为什么还能获取锁，是因为之前的线程调用了await方法，使其释放了锁。
             */
            readyLock.lock();
            try {
                if (!connected) {
                    connection();
                    //readied.await();
                    /**
                     * 给出一个阻塞超时时间，在zookeeper未更新服务器掉线状态前，发给这台服务器的消息都需要阻塞
                     * 所以在下线服务端前，一定要先降低服务器的权重，以减少打到这台机器上的消息量
                     * 缺点：如果某台服务器掉线了，在zookeeper更新状态前，会有大量消息打到这台掉线的机器上，
                     * 可能会导致很多消息超时3秒钟，这里是需要优化的逻辑点。
                     * 可以考虑在客户端启动的时候就先连上服务器，
                     * 这里用tryLock 的方式，作为客户端意外中断后的补救措施。
                     */
                    readied.await(Constants.DEFAULT_CONNECT_TIMEOUT, TimeUnit.MILLISECONDS);
                }
            } finally {
                readyLock.unlock();
            }
            //}
            if (!connected) {
                return null;
            }
        }

        DefaultFuture defaultFuture = null;//new DefaultFuture(request.getRequestId(), cChannel);
        if(request.getType() == RpcRequest.TYPE_ASYNC) {
            defaultFuture = new DefaultFuture(request.getRequestId(), cChannel,10*60*1000);
        }else {
            defaultFuture = new DefaultFuture(request.getRequestId(), cChannel);
        }
        Future future = channel.writeAndFlush(request);
        boolean success = future.await(Constants.DEFAULT_TIMEOUT);
        Throwable cause = future.cause();
        if (cause != null) {
            DefaultFuture.remove(request.getRequestId());
            throw new Exception("send faild by url:" + url.getAdress(), cause);
        }
        if (!success) {
            DefaultFuture.remove(request.getRequestId());
            throw new Exception("send faild by url:" + url.getAdress());
        }

        return defaultFuture.get();

    }

    /**
     * 结束工作线程，TODO：在清空客户端的时候调用
     */
    public synchronized void shutdown() {
        if (workerGroup != null) {
            workerGroup.shutdownGracefully();
        }
    }

    public void resetLasttime(){
        this.lastmsgtime = System.currentTimeMillis();
    }

    public boolean isConnected(){
        return connected;
    }

    public void tryitByheartbeat(){
        tryitnum ++;
        if(channel != null){
            RpcRequest request = new RpcRequest();
            request.setRequestId(UUID.randomUUID().toString());
            request.setType(RpcRequest.TYPE_HEARTBEAT);
            new DefaultFuture(request.getRequestId(), cChannel);
            try {
                channel.writeAndFlush(request);
            }catch (Exception e){
                throw  new RuntimeException(e);
            }
        }
    }
}
